package hbase;

import org.apache.hadoop.conf.Configuration;
//import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
//import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

//HBase Java API使用IDEA开发
//https://blog.csdn.net/jinxing_000/article/details/130039092

public class WriteHbase {

    public static String revertInt(int i){
        //rowid是8位纯数字
        String formatted = String.format("%08d", i);
        //倒序方式返回
        return new StringBuffer(formatted).reverse().toString();
    }
    public static void main(String[] args) {
        //设置日志路径
        String logDir = System.getProperty("log.dir");

        if (logDir == null) {
            // 开发环境中使用
            logDir = System.getProperty("user.dir")+"/logs/";
            System.setProperty("log.dir", logDir);
        }

        System.out.println("系统日志路径: " + System.getProperty("log.dir"));

        Configuration conf=null;
        Connection conn=null;
        Admin admin=null;
        conf= HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "node1:2181,node2:2181,node3:2181");
        //conf.addResource("core-site.xml");
        //conf.addResource("hbase-site.xml");

        try {
            conn = ConnectionFactory.createConnection(conf);
            //admin = conn.getAdmin();
            //long startTime = System.currentTimeMillis();
            String rowid;
            //批量插入，快
            List<Put> puts = new ArrayList<Put>();
            Table table = conn.getTable(TableName.valueOf("split_table_test"));
            int cnt=200002;
            int batch=2000;
            for (int id = 1; id <= cnt; id++) {
                rowid = revertInt(id);
                Put put = new Put(rowid.getBytes());
                //数字，日期都必须转为String
                put.addColumn("cf".getBytes(), "id".getBytes(), String.valueOf(id).getBytes());
                put.addColumn("cf".getBytes(), "name".getBytes(), String.format("name%d", id).getBytes());
                puts.add(put);
                //每batch行批量写入一次
                if((id%batch)==0 || id == cnt) {
                    System.out.println(String.format("正在写入第%d批次，运行查看区分使用情况：http://node1:60010/table.jsp?name=split_table_test",id/batch));
                    table.put(puts);
                }
            }
            table.close();
            System.out.println("数据插入成功！");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}